package ourvod.rtp;

import java.io.IOException;
import java.net.DatagramPacket;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.media.protocol.ContentDescriptor;
import javax.media.protocol.PushSourceStream;
import javax.media.protocol.SourceTransferHandler;
import javax.media.rtp.OutputDataStream;
import javax.media.rtp.RTPConnector;

import ourvod.jxta.P2PSystem;
import net.jxta.peergroup.PeerGroup;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.socket.JxtaMulticastSocket;

/**
 * RTPConnector base on JxtaMulticastSocket
 * 
 * @author cuizhenfu@gmail.com
 */
public class RTPJxtaMulticastSocketAdapter implements RTPConnector {
    
    private static final Logger logger = Logger.getLogger(RTPJxtaMulticastSocketAdapter.class.getName());

    JxtaMulticastSocket dataSock;
    JxtaMulticastSocket ctrlSock;
    SockInputStream dataInStream = null, ctrlInStream = null;
    SockOutputStream dataOutStream = null, ctrlOutStream = null;

    public RTPJxtaMulticastSocketAdapter(PipeAdvertisement dataPipeAdv, PipeAdvertisement ctrlPipeAdv) throws IOException {
        dataSock = new JxtaMulticastSocket(P2PSystem.getInstance().getOurvodPeerGroup(), dataPipeAdv);
        ctrlSock = new JxtaMulticastSocket(P2PSystem.getInstance().getOurvodPeerGroup(), ctrlPipeAdv);
    }

    /**
     * Returns an input stream to receive the RTP data.
     * 
     * @return
     * @throws IOException 
     */
    @Override
    public PushSourceStream getDataInputStream() throws IOException {
        if (dataInStream == null) {
            dataInStream = new SockInputStream(dataSock);
            dataInStream.start();
        }
        return dataInStream;
    }

    /**
     * Returns an output stream to send the RTP data.
     * 
     * @return
     * @throws IOException 
     */
    @Override
    public OutputDataStream getDataOutputStream() throws IOException {
        if (dataOutStream == null)
            dataOutStream = new SockOutputStream(dataSock);
        return dataOutStream;
    }

    /**
     * Returns an input stream to receive the RTCP data.
     * 
     * @return
     * @throws IOException 
     */
    @Override
    public PushSourceStream getControlInputStream() throws IOException {
        if (ctrlInStream == null) {
            ctrlInStream = new SockInputStream(ctrlSock);
            ctrlInStream.start();
        }
        return ctrlInStream;
    }

    /**
     * Returns an output stream to send the RTCP data.
     * 
     * @return
     * @throws IOException 
     */
    @Override
    public OutputDataStream getControlOutputStream() throws IOException {
        if (ctrlOutStream == null)
            ctrlOutStream = new SockOutputStream(ctrlSock);
        return ctrlOutStream;
    }

    /**
     * Close all the RTP, RTCP streams.
     */
    @Override
    public void close() {
        if (dataInStream != null)
            dataInStream.kill();
        if (ctrlInStream != null)
            ctrlInStream.kill();
        dataSock.close();
        ctrlSock.close();
    }

    /**
     * Set the receive buffer size of the RTP data channel.
     * This is only a hint to the implementation. The actual implementation
     * may not be able to do anything to this.
     * 
     * @param size
     * @throws IOException 
     */
    @Override
    public void setReceiveBufferSize(int size) throws IOException {
        dataSock.setReceiveBufferSize(size);
    }

    /**
     * Get the receive buffer size set on the RTP data channel.
     * Return -1 if the receive buffer size is not applicable for
     * the implementation.
     * 
     * @return 
     */
    @Override
    public int getReceiveBufferSize() {
        try {
            return dataSock.getReceiveBufferSize();
        } catch (Exception e) {
            logger.log(Level.WARNING, "Exception: xxx", e);
            return -1;
        }
    }

    /**
     * Set the send buffer size of the RTP data channel.
     * This is only a hint to the implementation. The actual implementation
     * may not be able to do anything to this.
     * 
     * @param size
     * @throws IOException 
     */
    @Override
    public void setSendBufferSize(int size) throws IOException {
        dataSock.setSendBufferSize(size);
    }

    /**
     * Get the send buffer size set on the RTP data channel.
     * Return -1 if the send buffer size is not applicable for
     * the implementation.
     * 
     * @return 
     */
    @Override
    public int getSendBufferSize() {
        try {
            return dataSock.getSendBufferSize();
        } catch (Exception e) {
            logger.log(Level.WARNING, "Exception: xxx", e);
            return -1;
        }
    }

    /**
     * Return the RTCP bandwidth fraction.  This value is used to
     * initialize the RTPManager.  Check RTPManager for more detauls.
     * Return -1 to use the default values.
     */
    @Override
    public double getRTCPBandwidthFraction() {
        return -1;
    }

    /**
     * Return the RTCP sender bandwidth fraction.  This value is used to
     * initialize the RTPManager.  Check RTPManager for more detauls.
     * Return -1 to use the default values.
     */
    @Override
    public double getRTCPSenderBandwidthFraction() {
        return -1;
    }

    /**
     * An inner class to implement an OutputDataStream based on UDP sockets.
     */
    class SockOutputStream implements OutputDataStream {

        JxtaMulticastSocket sock;

        public SockOutputStream(JxtaMulticastSocket sock) {
            this.sock = sock;
        }

        @Override
        public int write(byte data[], int offset, int len) {
            try {
                DatagramPacket packet = new DatagramPacket(data, offset, len);
                sock.send(packet);
            } catch (Exception e) {
                logger.log(Level.WARNING, "Exception: xxx", e);
                return -1;
            }
            return len;
        }
    }

    /**
     * An inner class to implement an PushSourceStream based on UDP sockets.
     */
    class SockInputStream extends Thread implements PushSourceStream {

        JxtaMulticastSocket sock;
        boolean done = false;
        boolean dataRead = false;
        SourceTransferHandler sth = null;

        public SockInputStream(JxtaMulticastSocket sock) {
            this.sock = sock;
        }

        /**
         * Read from the stream without blocking.
         * Returns -1 when the end of the media
         * is reached.
         *
         * @param buffer The buffer to read bytes into.
         * @param offset The offset into the buffer at which to begin writing data.
         * @param length The number of bytes to read.
         * @return The number of bytes read or -1
         * when the end of stream is reached.
         * @throws IOException Thrown if an error occurs while reading
         *
         */        
        @Override
        public int read(byte buffer[], int offset, int length) {
            byte[] buf = new byte[length];
            DatagramPacket p = new DatagramPacket(buf, length);
            try {
                sock.receive(p);
                byte[] data = p.getData();
                System.arraycopy(data, 0, buffer, offset, p.getLength());
            } catch (IOException e) {
                logger.log(Level.WARNING, "Exception: xxx", e);
                return -1;
            }
            synchronized (this) {
                dataRead = true;
                notify();
            }
            return p.getLength();
        }

        @Override
        public synchronized void start() {
            super.start();
            if (sth != null) {
                dataRead = true;
                notify();
            }
        }

        public synchronized void kill() {
            done = true;
            notify();
        }

        /**
         * Determine the size of the buffer needed for the data transfer.
         * This method is provided so that a transfer handler
         * can determine how much data, at a minimum, will be
         * available to transfer from the source.
         * Overflow and data loss is likely to occur if this much
         * data isn't read at transfer time.
         *
         * @return The size of the data transfer.
         */
        @Override
        public int getMinimumTransferSize() {
            return 2 * 1024;    // twice the MTU size, just to be safe.
        }

        /**
         * Register an object to service data transfers to this stream.
         * <p>
         * If a handler is already registered when
         * <CODE>setTransferHandler</CODE> is called,
         * the handler is replaced;
         * there can only be one handler at a time.
         * 
         * @param transferHandler The handler to transfer data to.
         */
        @Override
        public synchronized void setTransferHandler(SourceTransferHandler sth) {
            this.sth = sth;
            dataRead = true;
            notify();
        }

        // Not applicable.
        @Override
        public ContentDescriptor getContentDescriptor() {
            return null;
        }

        // Not applicable.
        @Override
        public long getContentLength() {
            return LENGTH_UNKNOWN;
        }

        // Not applicable.
        @Override
        public boolean endOfStream() {
            return false;
        }

        // Not applicable.
        @Override
        public Object[] getControls() {
            return new Object[0];
        }

        // Not applicable.
        @Override
        public Object getControl(String type) {
            return null;
        }

        /**
         * Loop and notify the transfer handler of new data.
         */
        @Override
        public void run() {
            while (!done) {

                synchronized (this) {
                    while (!dataRead && !done) {
                        try {
                            wait();
                        } catch (InterruptedException e) {
                        }
                    }
                    dataRead = false;
                }

                if (sth != null && !done) {
                    sth.transferData(this);
                }
            }
        }
    }
}
